package com.wanglei.rdd.transform

import org.apache.spark.{SparkConf, SparkContext}

object Spark18_combinebykey {

  def main(args: Array[String]): Unit = {

    //
    val conf = new SparkConf().setMaster("local[2]").setAppName("map")
    val sc = new SparkContext(conf)

    val rdd = sc.makeRDD(List(
      ("a", 1), ("a", 2), ("b", 3),
      ("b", 4), ("b", 5), ("a", 6)
    ), 2)

    // 计算同一个key的平均值
    // 需要三个参数
    // 第一个 将第一个数据进行结构转换 zeroValue
    // 第二个 分区内的计算规则
    // 第三个 分区间的计算规则
    rdd.combineByKey(
      v => (v, 1),
      (t: (Int, Int), v) => (t._1 + v, t._2 + 1),
      (t1: (Int, Int), t2: (Int, Int)) => (t1._1 + t2._1, t1._2 + t2._2)
    )
      .mapValues {
        case (sum, cnt) => sum / cnt
      }
      .collect().foreach(println)

    sc.stop()

  }

}


